package com.ruoyi.kafka;

import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @program: ruoyi
 * @description: KafkaAdminTest
 * @author: zengchen
 * @create: 2020-10-15 16:39
 **/
public class KafkaProducerTest {

    public static void main(String[] args) throws Exception {
        // 异步发送消息
//        asyncSend();

        // 同步发送消息 （异步任务get()阻塞）
//        syncSend();

        // 回调发送
//        callbackSend();

        // 主动分配分区发送
        partitionSend();
    }

    // 主动分配分区发送
    private static void partitionSend() {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaAdminTest.BOOTSTRAP_SERVERS_CONFIG);
        properties.put(ProducerConfig.ACKS_CONFIG, "all");//ProducerConfig.ACKS_CONFIG  源码有英文注释
        properties.put(ProducerConfig.RETRIES_CONFIG, "0");
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
        properties.put(ProducerConfig.LINGER_MS_CONFIG, "1");
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.ruoyi.kafka.StudyTestPartition");

        // Producer的主对象
        Producer<String, String> producer = new KafkaProducer<>(properties);
        // 消息对象 - ProducerRecorder
        for (int i = 0; i < 11; i++) {
            ProducerRecord<String, String> record =
                    new ProducerRecord<>(KafkaAdminTest.TOPIC_NAME, "callback-key-" + i, "callback-value-" + i);
            producer.send(record, (metadata, exception) -> {
                System.out.println("callback: " + ToStringBuilder.reflectionToString(metadata));
            });
        }
        // 所有的通道打开都需要关闭
        producer.close();
    }

    // 回调发送
    private static void callbackSend() throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaAdminTest.BOOTSTRAP_SERVERS_CONFIG);
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        properties.put(ProducerConfig.RETRIES_CONFIG, "0");
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
        properties.put(ProducerConfig.LINGER_MS_CONFIG, "1");
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // Producer的主对象
        Producer<String, String> producer = new KafkaProducer<>(properties);
        // 消息对象 - ProducerRecorder
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record =
                    new ProducerRecord<>(KafkaAdminTest.TOPIC_NAME, "callback-key-" + i, "callback-value-" + i);
            producer.send(record, (metadata, exception) -> {
                System.out.println("callback: " + ToStringBuilder.reflectionToString(metadata));
            });
        }
        // 所有的通道打开都需要关闭
        producer.close();
    }

    // 同步发送消息 （异步任务get()阻塞）
    private static void syncSend() throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaAdminTest.BOOTSTRAP_SERVERS_CONFIG);
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        properties.put(ProducerConfig.RETRIES_CONFIG, "0");
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
        properties.put(ProducerConfig.LINGER_MS_CONFIG, "1");
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // Producer的主对象
        Producer<String, String> producer = new KafkaProducer<>(properties);
        // 消息对象 - ProducerRecorder
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record =
                    new ProducerRecord<>(KafkaAdminTest.TOPIC_NAME, "sync-key-" + i, "sync-value-" + i);
            Future<RecordMetadata> send = producer.send(record);
            RecordMetadata recordMetadata = send.get();
            System.out.println(ToStringBuilder.reflectionToString(recordMetadata));
        }
        // 所有的通道打开都需要关闭
        producer.close();
    }

    // 异步发送消息
    private static void asyncSend() {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaAdminTest.BOOTSTRAP_SERVERS_CONFIG);
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        properties.put(ProducerConfig.RETRIES_CONFIG, "0");
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
        properties.put(ProducerConfig.LINGER_MS_CONFIG, "1");
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // Producer的主对象
        Producer<String, String> producer = new KafkaProducer<>(properties);
        // 消息对象 - ProducerRecorder
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record =
                    new ProducerRecord<>(KafkaAdminTest.TOPIC_NAME, "async-key-" + i, "async-value-" + i);
            producer.send(record);
        }
        // 所有的通道打开都需要关闭
        producer.close();
    }


}
